package me.pearjelly.pmc.service.impl;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import me.pearjelly.pmc.dao.DocDAO;
import me.pearjelly.pmc.service.InitialService;
import me.pearjelly.pmc.util.FileUtils;
import me.pearjelly.pmc.util.Logger;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.cn.smart.SmartChineseAnalyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.IndexWriter.MaxFieldLength;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;
import org.apache.lucene.util.Version;

public class InitialServiceMutiThreadImpl implements InitialService {
	private class IndexThread implements Runnable{
		private IndexWriter indexWriter;
		private DocDAO docDAO;
		private int startIndex;
		private int size;
		public IndexThread(IndexWriter indexWriter, DocDAO docDAO, int startIndex, int size) {
			this.indexWriter = indexWriter;
			this.docDAO = docDAO;
			this.startIndex = startIndex;
			this.size = size;
		}
		@Override
		public void run() {
			log.info("Starting index rowno:%d-%d.", startIndex, startIndex+size-1);
			List<Document> list = docDAO.selectCouldIndex(startIndex, size);
			int count = 0;
			for (Document document : list) {
				try {
					indexWriter.addDocument(document);
					count++;
				} catch (IOException e) {
					log.error("Index document:%s IOException:%s", document, e.getMessage());
				}
				log.debug("Index document:%s", document);
			}
			log.info("Ending, indexed %d documents.", count);
		}
	}

	@Override
	public void index() {
		IndexWriter indexWriter = this.prepareWriter();
		if(indexWriter != null){
			int total = docDAO.selectCouldIndexCount();
			int count = total%maxRows == 0?total/maxRows : total/maxRows + 1; 
			int i=0;
			
			for(;i<count;i++){
				executorService.submit(new IndexThread(indexWriter, docDAO, i*maxRows, maxRows));
			}
			executorService.shutdown();
			try {				
				boolean flag = executorService.awaitTermination(30, TimeUnit.MINUTES);
				if(flag){
					log.info("All IndexThread terminated.");
				}else{
					log.error("ExecutorService timeout.");
				}
			} catch (InterruptedException e) {
				log.error("ExecutorService InterruptedException:%s", e.getMessage());
			} finally {
				try {
					indexWriter.optimize();
					indexWriter.close();
				} catch (IOException e) {
					log.error("Close IndexWriter dir:%s IOException:%s", indexDir, e.getMessage());
				}
			}			
		}

	}
	
	/**
	 * 预处理索引目录。
	 * 如果存在历史数据，则删除。
	 * 如果目录不存在，则创建。
	 * @return
	 */
	public IndexWriter prepareWriter(){
		File dir = new File(this.indexDir);
		if(dir.exists()){
			FileUtils.deleteOnRecursion(dir);
		}
		try {
			Directory directory = FSDirectory.open(dir);
			IndexWriter indexWriter = new IndexWriter(directory, this.analyzer, MaxFieldLength.LIMITED);
			return indexWriter;
		} catch (IOException e) {
			log.error("Prepare IndexWriter dir:%s IOException:%s", indexDir, e.getMessage());
		}
		return null;
	}
	
	private static final Logger log = new Logger();
	private static final ExecutorService executorService = Executors.newFixedThreadPool(10);
	private DocDAO docDAO;
	private String indexDir = "/opt/pms/index/data";
	private Analyzer analyzer = new SmartChineseAnalyzer(Version.LUCENE_30);
	private int maxRows = 2000;

	public void setDocDAO(DocDAO docDAO) {
		this.docDAO = docDAO;
	}

	public void setIndexDir(String indexDir) {
		this.indexDir = indexDir;
	}

	public void setAnalyzer(Analyzer analyzer) {
		this.analyzer = analyzer;
	}

	public void setMaxRows(int maxRows) {
		this.maxRows = maxRows;
	}	
	
}
